package com.shujia.kafka

import org.apache.kafka.clients.consumer.{ConsumerRecord, ConsumerRecords, KafkaConsumer}

import java.util
import java.util.Properties

object Demo3KafkaConsumer {
  def main(args: Array[String]): Unit = {
    val properties = new Properties()
    //kafka 集群列表
    properties.setProperty("bootstrap.servers", "master:9092,node1:9092,node2:9092")

    //指定反序列化的方式
    //反序列化方式必须和序列化方式保存一致
    properties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
    properties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")

    //消费者组
    //同一条数据在一个组内只消费一次
    properties.setProperty("group.id", "aasdassdasdas")

    //禁用自动提交消费偏移量,默认是自动提交
    //消费偏移量保存再__consumer_offsets topic中
    properties.setProperty("enable.auto.commit", "true")

    //提交消费偏移量的间隔时间，默认5秒
    properties.setProperty("auto.commit.interval.ms", "1000")

    /**
     * earliest
     * 当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费
     * latest  默认
     * 当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产认值生的该分区下的数据
     * none
     * topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常
     *
     */
    //指定从什么位置开始消费
    properties.setProperty("auto.offset.reset", "earliest")

    //创建消费者
    val consumer = new KafkaConsumer[String, String](properties)

    val topics = new util.ArrayList[String]()
    topics.add("student")

    //订阅消费的topic
    consumer.subscribe(topics)

    while (true) {
      println("消费数据")

      //消费数据
      val consumerRecords: ConsumerRecords[String, String] = consumer.poll(1000)

      //获取消费的所有的数据
      val records: util.Iterator[ConsumerRecord[String, String]] = consumerRecords.iterator()

      //循环获取数据
      while (records.hasNext) {
        val record: ConsumerRecord[String, String] = records.next()

        val topic: String = record.topic()
        val partition: Int = record.partition()
        val offset: Long = record.offset()
        val timestamp: Long = record.timestamp()
        val key: String = record.key()
        val value: String = record.value()

        println(s"$topic\t$partition\t$offset\t$timestamp\t$key\t$value")
      }
    }


    consumer.close()

  }

}
